import json
import re

from mrjob.job import MRJob
from mrjob.step import MRStep


class MyMrjob(MRJob):
    def steps(self):
        return [MRStep(mapper=self.mymapper,
                       reducer=self.mymapper
                       ),
                MRStep(mapper=self.mymapper2,
                       reducer=self.myreducer2
                       )
                ]

    def mymapper(self, key, value):
        try:
            job_info = json.loads(value)
            address = job_info['address']
            m = re.match('(?P<min_salary>[\d\.]+)-(?P<max_salary>[0-9\.]+)(?P<unit>(万|千))/(?P<time>(月|年))',
                         job_info['salary'])
            if not m:
                return
            min_salary = float(m.group('min_salary'))
            max_salary = float(m.group('max_salary'))

            if m.group('unit') == '万':
                min_salary *= 10
                max_salary *= 10
            if m.group('time') == '年':
                min_salary /= 12
                max_salary /= 12

            avg_salary = (min_salary + max_salary) / 2
            yield (address, avg_salary)
        except Exception as e:
            pass

    def myreducer(self, key, values):
        values = list(values)
        avg_salary = sum(values) / len(values)
        yield (key, (avg_salary, len(values)))

    def mymapper2(self, key, values):
        yield (None, (key, values))

    def myreducer2(self, key, values):
        values = sorted(list(values), key=lambda e: e[1][1])[-3:]
        for value in values:
            yield value


if __name__ == '__main__':
    import os

    os.system('rm -rf ./result')
    MyMrjob.run()

    for file in os.listdir('./result'):
        with open('./result/' + file,  'r') as f:
            while True:
                line = f.readline()
                if not line:
                    break
                key, value = line.split('\t')
                print('{}/t{}'.format(json.loads(key), json.loads(value)))
